package com.adcc.loadsheet.monitor.util;

import com.adcc.aoc.devLib.common.log.Log;
import com.adcc.aoc.devLib.traffic.base.ConnectState;
import com.adcc.aoc.devLib.traffic.base.channel.DataEventListener;
import com.adcc.aoc.devLib.traffic.base.channel.StateEventListener;
import com.adcc.aoc.devLib.traffic.base.packet.DataPacket;
import com.adcc.aoc.devLib.traffic.base.udp.UDPConfiguration;
import com.adcc.aoc.devLib.traffic.base.udp.UDPServer;
import com.adcc.aoc.devLib.traffic.base.udp.UDPTrafficMode;
import com.adcc.loadsheet.monitor.congifuration.AppConfiguration;
import com.adcc.loadsheet.monitor.entity.MonitorStatus;
import com.adcc.loadsheet.monitor.entity.MsgUpCenterMonitor;
import com.adcc.loadsheet.monitor.entity.UDPConfig;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Queues;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.Minutes;
import org.joda.time.format.DateTimeFormat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

@Component
public class MsgUpCenterUDPFactory extends BaseUDPFactory implements DataEventListener, StateEventListener {

    @Autowired
    private AppConfiguration appConfiguration;

    @Autowired
    private SystemConfigFactory systemConfigFactory;

    // NTP告警时间阀值
    private int ntpAlarmTime;

    private int timespan;

    private int points;

    private long updateTime;

    private AtomicInteger count = new AtomicInteger(1);

    private AtomicLong loadsheetNum = new AtomicLong(0L);

    private AtomicLong passgerlistNum = new AtomicLong(0L);

    private ArrayBlockingQueue<Long> loadsheetData;

    private ArrayBlockingQueue<Long> passgerlistData;

    // MsgUpCenter监控信息
    private MsgUpCenterMonitor monitor = new MsgUpCenterMonitor();

    public MsgUpCenterUDPFactory() {

    }

    public MsgUpCenterUDPFactory(String localHost, String remoteHost, int port, UDPTrafficMode trafficMode, int readerIdle) {
        this.localHost = localHost;
        this.remoteHost = remoteHost;
        this.port = port;
        this.trafficMode = trafficMode;
        this.readerIdle = readerIdle;
    }

    @PostConstruct
    private void initialize() {
        try {
            localHost = appConfiguration.getHost();
            if(systemConfigFactory.getMsgUpCenterConfigMap().containsKey("udp")){
                // 加载配置
                loadMsgUpCenterConfig(systemConfigFactory.getMsgUpCenterConfigMap());
                // 启动
                start();
            }
        } catch (Exception ex) {
            Log.error(MsgUpCenterUDPFactory.class, "initialize() error", ex);
        }
    }

    public MsgUpCenterMonitor getMonitor() {
        return monitor;
    }

    public void setMonitor(MsgUpCenterMonitor monitor) {
        this.monitor = monitor;
    }

    /**
     * 初始化队列数据
     */
    private void initQueueData() {

        // 初始化数据队列
        loadsheetData = Queues.newArrayBlockingQueue(points);
        while(loadsheetData.size() < points) {
            loadsheetData.offer(0L);
        }
        passgerlistData = Queues.newArrayBlockingQueue(points);
        while(passgerlistData.size() < points) {
            passgerlistData.offer(0L);
        }
    }

    /**
     * 加载MsgUpCenter配置信息
     * @param msgUpCenterConfigMap
     */
    public void loadMsgUpCenterConfig(Map<String, Object> msgUpCenterConfigMap) {
        try {
            // 读取UDP配置
            if(msgUpCenterConfigMap.containsKey("udp")) {
                UDPConfig udpConfig = (UDPConfig) msgUpCenterConfigMap.get("udp");
                remoteHost = udpConfig.getHost();
                port = Integer.valueOf(udpConfig.getPort());
                readerIdle = Integer.valueOf(udpConfig.getTimeout());
            }

            // 读取NTP配置
            if(msgUpCenterConfigMap.containsKey("ntp")) {
                Map<String, String> ntpMap = (Map<String, String>) msgUpCenterConfigMap.get("ntp");
                ntpAlarmTime = ntpMap.containsKey("alarmTime") ? Integer.valueOf(ntpMap.get("alarmTime")) : 0;
            }

            if(msgUpCenterConfigMap.containsKey("lineChart")) {
                Map<String, String> lineChartMap = (Map<String, String>) msgUpCenterConfigMap.get("lineChart");
                points = Integer.valueOf(lineChartMap.get("points"));
                timespan = Integer.valueOf(lineChartMap.get("timespan"));
            }

            // 设置当前监控信息
            monitor.setLocalDateTime(DateTime.now(DateTimeZone.UTC).toString("yyyy-MM-dd HH:mm:ss"));
            monitor.setTimespan(timespan);
            monitor.setPoints(points);

            synchronized (this) {
                // 初始化数据队列
                initQueueData();
                loadsheetNum.set(0L);
                passgerlistNum.set(0L);
                count.set(1);
                updateTime = DateTime.now(DateTimeZone.UTC).getMillis();
                monitor.setUpdateTime(updateTime);
                monitor.setLoadsheetData(loadsheetData);
                monitor.setPassgerlistData(passgerlistData);
            }

        } catch (Exception ex) {
            Log.error(MsgUpCenterUDPFactory.class, "loadMsgUpCenterConfig() error", ex);
        }
    }

    public void start() {
        UDPConfiguration u = new UDPConfiguration();
        u.setLocalHost(localHost);
        u.setRemoteHost(remoteHost);
        u.setPort(port);
        u.setWorkerGroup(5);
        u.setTrafficMode(trafficMode);
        u.setRcvBuf(1024*128);
        u.setReaderIdle(readerIdle);
        u.setWriterIdle(0);
        u.setAllIdle(0);
        server = new UDPServer(u);
        server.setDataEventListener(this);
        server.setStateEventListener(this);
        server.start();
    }

    public void stop() {
        if(server != null) {
            server.stop();
        }
    }

    public void restart() {
        stop();
        start();
    }

    /**
     * 错误的监控信息
     */
    private void errorStatus() {
        monitor.setNtpStatus(false);
        monitor.setUplinkStatus(false);
        monitor.setDownlinkStatus(false);
        monitor.setStatus(MonitorStatus.ABNORMAL);
        // 本地时间（NTP时间）
        monitor.setLocalDateTime(DateTime.now(DateTimeZone.UTC).toString("yyyy-MM-dd HH:mm:ss"));
        monitor.setRemoteDateTime("");
        monitor.setMinutesBetween(0);
        synchronized (this) {
            // 初始化数据队列
            initQueueData();
            loadsheetNum.set(0L);
            passgerlistNum.set(0L);
            count.set(1);
            updateTime = DateTime.now(DateTimeZone.UTC).getMillis();
            monitor.setUpdateTime(updateTime);
            monitor.setLoadsheetData(loadsheetData);
            monitor.setPassgerlistData(passgerlistData);
        }
    }

    @Override
    public void onData(DataPacket packet) {
        try {
            // 避免多个UDP使用同一个接口发送数据，造成接收数据混乱
            if(remoteHost.equals(packet.getHead().get("remoteAddress"))){
                String msg = new String(packet.getData());
                Log.info(MsgUpCenterUDPFactory.class, "receive msgUpCenter udp msg: " + msg);
                JSONObject jsonObject = JSON.parseObject(msg);

                // 本地时间（NTP时间）
                monitor.setLocalDateTime(DateTime.now(DateTimeZone.UTC).toString("yyyy-MM-dd HH:mm:ss"));

                // 报文总数
                if(jsonObject.containsKey("Total")){
                    monitor.setTotalCount(jsonObject.getLong("Total"));
                }

                // 舱单数
                if(jsonObject.containsKey("LoadsheetNum")){
                    monitor.setLoadsheetCount(jsonObject.getLong("LoadsheetNum"));
                }

                // 名单数
                if(jsonObject.containsKey("PassgerlistNum")){
                    monitor.setPassgerlistCount(jsonObject.getLong("PassgerlistNum"));
                }

                // 失败数
                if(jsonObject.containsKey("Fail")){
                    monitor.setFailCount(jsonObject.getLong("Fail"));
                }

                // 上行状态（ACARS接收）
                if(jsonObject.containsKey("Uplink")){
                    monitor.setUplinkStatus("UP".equalsIgnoreCase(jsonObject.getString("Uplink")));
                }

                // 下行状态（ACARS发送）
                if(jsonObject.containsKey("Downlink")){
                    monitor.setDownlinkStatus("UP".equalsIgnoreCase(jsonObject.getString("Downlink")));
                }

                // UDP发送时间，即远程时间（运行程序的本地时间）
                if(jsonObject.containsKey("sendTime")){
                    String timestr=jsonObject.getString("sendTime");
                    if(timestr.contains("/")){
                        timestr=DateTime.parse(timestr, DateTimeFormat.forPattern("yyyy/MM/dd HH:mm:ss")).toString("yyyy-MM-dd HH:mm:ss");
                    }
                    monitor.setRemoteDateTime(timestr);
                }

                // 进行本地时间和远程时间对比，计算时间差(远程时间-本地时间)
                int minutesBetween = Minutes.minutesBetween(DateTime.parse(monitor.getLocalDateTime(), DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")), DateTime.parse(monitor.getRemoteDateTime(), DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss"))).getMinutes();
                monitor.setMinutesBetween(minutesBetween);

                // 进行NTP状态判断（依据时间差） 告警时间阀值为0时，状态为true
                monitor.setNtpStatus(ntpAlarmTime == 0 || Math.abs(monitor.getMinutesBetween()) < ntpAlarmTime);

                synchronized (this) {
                    // UDP消息每60s发送一次
                    // 在一个时间间隔时间内，统计一次舱单/名单数据
                    if(count.intValue() <= timespan){
                        loadsheetNum.addAndGet(monitor.getLoadsheetCount());
                        passgerlistNum.addAndGet(monitor.getPassgerlistCount());
                        // 第五次计数时，将五分钟内的舱单/名单数据，存入对应队列中
                        if(count.intValue() == timespan){
                            // 队列数到达当前队列容量时，poll
                            if(loadsheetData.size() == points){
                                loadsheetData.poll();
                            }
                            loadsheetData.offer(loadsheetNum.longValue());

                            if(passgerlistData.size() == points){
                                passgerlistData.poll();
                            }
                            passgerlistData.offer(passgerlistNum.longValue());
                            // 监控信息存入舱单/名单数据
                            monitor.setLoadsheetData(loadsheetData);
                            monitor.setPassgerlistData(passgerlistData);
                            // 设置当前更新时间
                            updateTime = DateTime.now(DateTimeZone.UTC).getMillis();
                            monitor.setUpdateTime(updateTime);
                            // 初始化计数
                            loadsheetNum.set(0L);
                            passgerlistNum.set(0L);
                            count.set(1);
                        }else{
                            count.incrementAndGet();
                        }
                    }
                }

                // 进行监控状态判断
                if(!monitor.isNtpStatus() || !monitor.isUplinkStatus() || !monitor.isDownlinkStatus()){
                    monitor.setStatus(MonitorStatus.ABNORMAL);
                }else{
                    monitor.setStatus(MonitorStatus.NORMAL);
                }
            }
        } catch (Exception ex) {
            errorStatus();
            Log.error(MsgUpCenterUDPFactory.class, "onData() error", ex);
        }
    }

    @Override
    public void onState(String channelID, String remoteAddress, ConnectState state) {
        try {
            if(state != ConnectState.CONNECTED){
                // 超时空闲或者接收异常时，所有状态都为异常
                errorStatus();
            }
        } catch (Exception ex) {
            errorStatus();
            Log.error(MsgUpCenterUDPFactory.class, "onState() error", ex);
        }
    }
}
